/** @file    CommunicationManagerParser.cpp
 *  @author  Alessandro Polo
 *  @version $Id: CommunicationManagerParser.cpp 2821 2010-08-01 11:44:26Z alex $
 *  @brief
 * File containing methods for the CommunicationManagerParser class.
 * The header for this class can be found in CommunicationManagerParser.h, check that file
 * for class description.
 ****************************************************************************/
/* Copyright (c) 2007-2010, WOSH - Wide Open Smart Home 
 * by Alessandro Polo - OpenSmartHome.com
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the OpenSmartHome.com WOSH nor the
 *       names of its contributors may be used to endorse or promote products
 *       derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY Alessandro Polo ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL Alessandro Polo BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 ****************************************************************************/

 #include "CommunicationManagerParser.h"

 #include <core/Message.h>
 #include <core/Request.h>
 #include <core/Response.h>
 #include <core/Fact.h>
 #include <core/MethodsCommon.h>
 #include <core/List.h>
 #include <core/Table.h>
 #include <core/Bundle.h>
 #include <core/NetworkManager.h>
 #include <core/SecurityManager.h>


using namespace std;
using namespace wosh;

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS

CommunicationManagerParser::CommunicationManagerParser( BundleGeneric& bundle, MapT<string, wosh::Communicator*>& communicators, MapT<string, wosh::Interpreter*>& interpreters )
	: BundleGenericWorker(bundle),  comms(communicators), transl(interpreters)
 {
	this->timeoutEval = 0;
}

CommunicationManagerParser::~CommunicationManagerParser() {
	if ( isThreadRunning() ) {
		quitThread(10000);
	 }
 }

//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////// RUN THREAD

void CommunicationManagerParser::runThread()
 {
	Log(LOG_VERBOSE, ":runThread(): STARTING" );

	long counterEval = 0;
	while( this->running )
	 {
		ThreadImpl::sleepForMSec(1000);
		setThreadAlive();
		if ( this->timeoutEval == 0 ) // disabled
			continue;

		++counterEval;
		if ( counterEval < this->timeoutEval )
			continue;
		counterEval = 0;
		evalTimeOut(); // remove timed-out comms/interp

		// update them all
		updateCommunicators();
		updateInterpreters();
	 }

	Log(LOG_VERBOSE, ":runThread(): EXITING" );
 }

////////////////////////////////////////////////////////////////////////////////////////////////// RUN THREAD
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

void CommunicationManagerParser::evalTimeOut()
 {
	Log(LOG_VERBOSE, ":evalTimeOut(): Evaluating Timedout Communicators/Interpreters.." );

	// check for timed out communicators
	int rem_c = 0;
	this->comms.transactionBeginWrite();
	tStrCommunicatorMap::Iterator it_c;
	const tStrCommunicatorMap::ConstIterator it_c_end = this->comms.end();
	for ( it_c=this->comms.begin(); it_c!=it_c_end; ++it_c ) {
		if ( it_c->second == NULL ) continue;
		if ( Utilities::std_time() - it_c->second->getSeen() > this->timeoutEval*1.5 ) {
			removeCommunicator_(it_c->second);
			this->comms.erase(it_c, false);
			++rem_c;
		 }
	 }
	this->comms.transactionEnd();

	// check for timed out interpreters
	int rem_i = 0;
	this->transl.transactionBeginWrite();
	tStrInterpreterMap::Iterator it_i;
	const tStrInterpreterMap::ConstIterator it_i_end = this->transl.end();
	for ( it_i=this->transl.begin(); it_i!=it_i_end; ++it_i ) {
		if ( it_i->second == NULL ) continue;
		if ( Utilities::std_time() - it_i->second->getSeen() > this->timeoutEval*1.5 ) {
			removeInterpreter_(it_i->second);
			this->transl.erase(it_i, false);
			++rem_i;
		 }
	 }
	this->transl.transactionEnd();

	if ( rem_c  == 0 && rem_i == 0 )
		return;
	Log(LOG_INFO, ":evalTimeOut(): Removed Timedout Communicators[%d] / Interpreters[%d]", rem_c, rem_i );
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

void CommunicationManagerParser::removeByKernel( const string& kernelName )
 {
	Log(LOG_VERBOSE, ":removeByKernel(%s): Removing Communicators/Interpreters..", kernelName.c_str() );

	int rem_c = 0;
	this->comms.transactionBeginWrite();
	tStrCommunicatorMap::Iterator it_c;
	const tStrCommunicatorMap::ConstIterator it_c_end = this->comms.end();
	for ( it_c=this->comms.begin(); it_c!=it_c_end; ++it_c ) {
		if ( it_c->second == NULL ) continue;
		if ( it_c->second->getURI().getKernelName() != kernelName ) continue;
		removeCommunicator_(it_c->second);
		this->comms.erase(it_c, false);
		++rem_c;
	 }
	this->comms.transactionEnd();

	int rem_i = 0;
	this->transl.transactionBeginWrite();
	tStrInterpreterMap::Iterator it_i;
	const tStrInterpreterMap::ConstIterator it_i_end = this->transl.end();
	for ( it_i=this->transl.begin(); it_i!=it_i_end; ++it_i ) {
		if ( it_i->second == NULL ) continue;
		if ( it_i->second->getURI().getKernelName() != kernelName ) continue;
		removeInterpreter_(it_i->second);
		this->transl.erase(it_i, false);
		++rem_i;
	 }
	this->transl.transactionEnd();

	if ( rem_c  == 0 && rem_i == 0 )
		return;
	Log(LOG_INFO, ":removeByKernel(%s): Removed Communicators[%d] / Interpreters[%d]", kernelName.c_str(), rem_c, rem_i );
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

void CommunicationManagerParser::removeCommunicator_( Communicator* comm )
 {
	if ( comm == NULL ) return;
	Log(LOG_INFO, ":removeCommunicator_(%s)", comm->getName().c_str() );
	delete comm;
 }

void CommunicationManagerParser::removeInterpreter_( Interpreter* inter )
 {
	if ( inter == NULL ) return;
	Log(LOG_INFO, ":removeInterpreter_(%s)", inter->getName().c_str() );
	delete inter;
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

WRESULT CommunicationManagerParser::updateCommunicators()
 {
	Log(LOG_INFO, ":updateCommunicators()" );

	return updateObjects( "*", _CommunicatorService_TYPE );
 }
 
WRESULT CommunicationManagerParser::updateInterpreters()
 {
	Log(LOG_INFO, ":updateInterpreters()" );

	return updateObjects( "*", _InterpreterService_TYPE );
 }

WRESULT CommunicationManagerParser::updateObjects( const std::string& kernel, const std::string& type )
 {
	Log(LOG_VERBOSE, ":updateObjects(%s) : Sending request.. [%s]", type.c_str(), kernel.c_str() );

	Request* request = new Request(_METHOD_ListProperties);
	Message* message = new Message(request);
	message->setDestinationBus( _Bus_Core );
	message->getDestination().fromString("*:**/*["_KEY_interfaces"~" + type + "]");
	message->getDestination().setKernelName(kernel);
	message->setSource( this->getOwner() );
	SecurityManager::getCredentialImplicit().signMessage(message, this->getOwner());
	BusCore.postMessage(message);
	return WRET_OK;
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

void CommunicationManagerParser::evalMessage( const Message& message )
 {
	if ( message.isEmpty() ) return;

	if ( message.getContent()->isFact() ) {
		const wosh::Fact* fact = message.getContent()->asFact();
		if ( message.getSource().getName() == "Network" && fact->hasData() ) {
			if ( fact->getEvent() == _WoshHost_EVENT_ListChanged ) {
				const WoshHost_ListChanged* wh_lc = message.getContent()->as<WoshHost_ListChanged>();
				if ( wh_lc == NULL ) return;
				Log(LOG_VERBOSE, ":evalMessage() : Host updated/added, Updating Communicators.." );
				updateObjects( wh_lc->getWoshHostName(), _CommunicatorService_TYPE );
				updateObjects( wh_lc->getWoshHostName(), _InterpreterService_TYPE );
			}
			else if ( fact->getEvent() == _WoshHost_EVENT_ConnectionChanged ) {
				const WoshHost_ConnectionChanged* wh_cc = message.getContent()->as<WoshHost_ConnectionChanged>();
				if ( wh_cc == NULL ) return;
				Log(LOG_VERBOSE, ":evalMessage() : Host disconnected/removed, Disabling Communicators.." );
				if ( wh_cc->getConnectionState() == 2 )
					removeByKernel( wh_cc->getWoshHostName() );
			}
		 }
	 }

	else if ( message.getContent()->isResponse() ) {
		const Response* response = message.getContent()->asResponse();
		if ( !response->hasData() ) return;
		if ( response->getMethod() != _METHOD_ListProperties )
			return;

		if ( !response->getData()->isKindOf<Table>() ) return;
		const wosh::Table* props = response->getData()->as<Table>();
		Variant vBundleInterface(_KEY_interfaces);
		int keyCol=0; int valueCol=1;
		int type = 0;
		const ISerializable*  iBundleInterfaces = props->findMappedColumnItem(keyCol, valueCol, &vBundleInterface);
		if ( iBundleInterfaces == NULL ) return;
		if ( iBundleInterfaces->isKindOf<Variant>() ) {
			const Variant* vBundleInterfaces = dynamic_cast<const Variant*>(iBundleInterfaces);
			if ( vBundleInterfaces == NULL ) return;
			if ( vBundleInterfaces->toString().find(_CommunicatorService_TYPE) != string::npos )
				type = 1;
			else if ( vBundleInterfaces->toString().find(_InterpreterService_TYPE) != string::npos )
				type = 2;
		 }
		else if ( iBundleInterfaces->isKindOf<List>() ) {
			const List* lBundleInterfaces = dynamic_cast<const List*>(iBundleInterfaces);
			if ( lBundleInterfaces == NULL ) return;
			if ( lBundleInterfaces->findItem(_CommunicatorService_TYPE) != -1 )
				type = 1;
			else if ( lBundleInterfaces->findItem(_InterpreterService_TYPE) != -1 )
				type = 2;
		 }

		if ( type == 1 ) {
			this->comms.transactionBeginWrite();
			wosh::Communicator* comm = this->comms.find( message.getSource().toString() );
			if ( comm == NULL ) {
				Log(LOG_INFO, ":evalMessage() : New Communicator found [%s]", message.getSource().toString().c_str() );
				comm = new wosh::Communicator();
				comm->setName( message.getSource().getName() );
				comm->setURI( message.getSource() );
				this->comms.set(comm->getURI().toString(), comm);
			 }
			else
				Log(LOG_INFO, ":evalMessage() : Updating Communicator [%s]", message.getSource().toString().c_str() );

			comm->setSeen();
			comm->getInterfaces().fromString( iBundleInterfaces->toString() );

			comm->setCost( props->findMappedColumnItem(keyCol, valueCol, _CommunicatorService_KEY_Cost).toDouble(0.0) );
			comm->setPriority( props->findMappedColumnItem(keyCol, valueCol, _CommunicatorService_KEY_Priority).toInteger(0) );
			comm->setReliability( props->findMappedColumnItem(keyCol, valueCol, _CommunicatorService_KEY_Reliability).toInteger(0) );
			comm->setChannelTypes( props->findMappedColumnItem(keyCol, valueCol, _CommunicatorService_KEY_ChannelType).toInteger(0) );
			comm->setChannelLink( (Communicator::CHANNEL_LINK)props->findMappedColumnItem(keyCol, valueCol, _CommunicatorService_KEY_ChannelLink).toInteger(0) );
			comm->setProtocol( (Notification::PROTOCOL)props->findMappedColumnItem(keyCol, valueCol, _CommunicatorService_KEY_Protocol).toInteger(0) );
			std::string locations = props->findMappedColumnItem(keyCol, valueCol, _CommunicatorService_KEY_Locations).toString();
			Utilities::splitString(locations, ";", comm->getLocations(), false, true);
			this->comms.transactionEnd();
		 }
		else if ( type == 2 ) {
			this->transl.transactionBeginWrite();
			wosh::Interpreter* inter = this->transl.find( message.getSource().toString() );
			if ( inter == NULL ) {
				Log(LOG_INFO, ":evalMessage() : New Interpreter found [%s]", message.getSource().toString().c_str() );
				inter = new wosh::Interpreter();
				inter->setName( message.getSource().getName() );
				inter->setURI( message.getSource() );
				this->transl.set(inter->getURI().toString(), inter);
			 }
			else
				Log(LOG_VERBOSE, ":evalMessage() : Updating Interpreter [%s]", message.getSource().toString().c_str() );

			inter->setSeen();
			inter->getInterfaces().fromString( iBundleInterfaces->toString() );

			inter->setInputTypes( props->findMappedColumnItem(keyCol, valueCol, _InterpreterService_KEY_InputTypes).toInteger(0) );
			inter->setOutputTypes( props->findMappedColumnItem(keyCol, valueCol, _InterpreterService_KEY_OutputTypes).toInteger(0) );
			inter->setInputProtocols( props->findMappedColumnItem(keyCol, valueCol, _InterpreterService_KEY_InputProtocols).toInteger(0) );
			inter->setOutputProtocols( props->findMappedColumnItem(keyCol, valueCol, _InterpreterService_KEY_OutputProtocols).toInteger(0) );

			std::string in_lang = props->findMappedColumnItem(keyCol, valueCol, _InterpreterService_KEY_InputLanguages).toString();
			Utilities::splitString(in_lang, ";", inter->getInputLanguages(), false, true);

			std::string out_lang = props->findMappedColumnItem(keyCol, valueCol, _InterpreterService_KEY_OutputLanguages).toString();
			Utilities::splitString(out_lang, ";", inter->getOutputLanguages(), false, true);

			this->transl.transactionEnd();
		 }
		else {
			Log(LOG_WARNING, ":evalMessage() : Unrecognised service [%s]", message.getSource().toString().c_str() );
		 }
	 }

 }
